package david.java.flink.join;

import david.java.flink.join.dataSource.ProDataSource;
import david.java.flink.join.dataSource.TxDataSource;
import david.java.flink.join.entity.JoinDate;
import david.java.flink.join.entity.Promotion;
import david.java.flink.join.entity.Tx;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.io.FileNotFoundException;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

/**
 * @Description:
 * @Author: David
 * @Date: Create in 10:43 上午 2021/11/8
 */
public class JoinTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        // DataStreamSource<Tx> tx = env.addSource(new TxDataSource());
        // DataStreamSource<Promotion> pro = env.addSource(new ProDataSource());

        // DataStreamSource<Tx> tx = env.addSource(new TxDataSource())
        //         .assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(1000)).withTimestampAssigner())
        //         .assignTimestampsAndWatermarks(WatermarkStrategy.noWatermarks().withTimestampAssigner((TimestampAssignerSupplier<Tx>)(t) -> t.get ).extractTimestamp())


        // tx.print();
        // pro.print();


        // window join
        // DataStream<JoinDate> apply = tx.join(pro)
        //         .where((KeySelector<Tx, Integer>) value -> value.getId())
        //         .equalTo((KeySelector<Promotion, Integer>) value -> value.getId())
        //         .window(TumblingProcessingTimeWindows.of(Time.seconds(2)))
        //         .apply((JoinFunction<Tx, Promotion, JoinDate>) (tx1, pro1) -> new JoinDate(tx1.getId(), tx1.getName(), pro1.getName(), tx1.getPrice(), pro1.getProPrice()));
        //
        //
        // apply.print();


        // interval join    Interval Join目前只支持Event Time。
        // SingleOutputStreamOperator<JoinDate> process = tx
        //         .keyBy((KeySelector<Tx, Integer>) (t) -> t.getId())
        //         .intervalJoin(pro.keyBy((KeySelector<Promotion, Integer>) (t) -> t.getId()))
        //         .between(Time.seconds(-1), Time.seconds(1))
        //         .process(new ProcessJoinFunction<Tx, Promotion, JoinDate>() {
        //             @Override
        //             public void processElement(Tx tx, Promotion pro, Context ctx, Collector<JoinDate> out) throws Exception {
        //                 out.collect(new JoinDate(tx.getId(), tx.getName(), pro.getName(), tx.getPrice(), pro.getProPrice()));
        //             }
        //         });
        //
        // process.print();


        // cogroup join
        // DataStream<JoinDate> coGroup = tx.coGroup(pro)
        //         .where((KeySelector<Tx, Integer>) Tx::getId)
        //         .equalTo((KeySelector<Promotion, Integer>) Promotion::getId)
        //         .window(TumblingProcessingTimeWindows.of(Time.seconds(3)))
        //         .apply(new CoGroupFunction<Tx, Promotion, JoinDate>() {
        //             @Override
        //             public void coGroup(Iterable<Tx> tx, Iterable<Promotion> pro, Collector<JoinDate> out) throws Exception {
        //               Map<Integer, Promotion> proMap = new HashMap<>();
        //               Map<Integer, Promotion> txMap = new HashMap<>();
        //               pro.forEach(t -> proMap.put(t.getId(), t));
        //               tx.forEach(t -> {
        //                   JoinDate joinDate = new JoinDate();
        //                   joinDate.setId(t.getId());
        //                   joinDate.setTxName(t.getName());
        //                   joinDate.setTxPrice(t.getPrice());
        //                   if (proMap.containsKey(t.getId())) {
        //                       joinDate.setProName(proMap.get(t.getId()).getName());
        //                       joinDate.setProPrice(proMap.get(t.getId()).getProPrice());
        //                   }
        //                   out.collect(joinDate);
        //               });
        //             }
        //         });
        //
        //     coGroup.print();









        env.execute("test join");

    }

}
